package com.javaDemo.thread;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import com.google.common.collect.Lists;

/**
 * 秒杀系统示例
 * 演示高并发下的库存扣减、请求合并、超时处理和回滚机制
 * 
 * 系统设计要点：
 * 1. 请求合并（Request Merging）：将多个用户请求合并为一个批次处理，减少数据库操作次数
 * 2. 库存预占（Stock Pre-allocation）：先预占库存，再异步处理订单
 * 3. 超时处理（Timeout Handling）：对超时请求进行回滚，避免库存死锁
 * 4. 并发控制（Concurrency Control）：使用同步机制确保库存操作的原子性
 * 5. 异常处理（Exception Handling）：处理各种异常情况，保证系统稳定性
 */
public class KillDemo {
    /**
     * 秒杀系统主流程
     * 1. 启动10个用户线程模拟并发请求
     * 2. 库存总数为6个
     * 3. 生成一个合并队列，每3个用户请求作为一个批次处理
     * 4. 每个用户能拿到自己的请求响应
     * 5. 对超时请求进行回滚处理
     * 
     * 注意：该示例主要演示秒杀系统的核心逻辑，实际生产环境中还需考虑：
     * - 分布式锁：确保多节点部署时的库存一致性
     * - 消息队列：异步处理订单，提高系统吞吐量
     * - 限流措施：防止系统被过多请求压垮
     * - 缓存预热：提前加载热点数据到缓存
     */
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        KillDemo killDemo = new KillDemo();
        killDemo.mergeJob();
        Thread.sleep(2000);

        CountDownLatch countDownLatch = new CountDownLatch(10);

        System.out.println("-------- 库存 --------");
        System.out.println("库存初始数量 :" + killDemo.stock);

        Map<UserRequest, java.util.concurrent.Future<Result>> requestFutureMap = new HashMap<>();
        for (int i = 0; i < 10; i++) {
            final Long orderId = i + 100L;
            final Long userId = Long.valueOf(i);
            UserRequest userRequest = new UserRequest(orderId, userId, 1);
            java.util.concurrent.Future<Result> future = executorService.submit(() -> {
                countDownLatch.countDown();
                countDownLatch.await(1, TimeUnit.SECONDS);
                return killDemo.operate(userRequest);
            });

            requestFutureMap.put(userRequest, future);
        }

        System.out.println("------- 客户端响应 -------");
        // 等待一段时间，让请求有足够时间被处理
        Thread.sleep(1000);
        // 获取并处理所有用户请求的结果
        requestFutureMap.entrySet().forEach(entry -> {
            try {
                // 设置获取结果的超时时间为300毫秒
                // 注意：这里可能存在潜在问题 - 如果获取结果超时，不代表请求处理失败，
                // 可能是请求仍在处理中，此时不应该立即回滚
                Result result = entry.getValue().get(300, TimeUnit.MILLISECONDS);
                System.out.println(Thread.currentThread().getName() + ":客户端请求响应:" + result);

                // 检查是否需要回滚 - 只有确认是等待超时的请求才回滚
                if (!result.isSuccess() && result.getMsg().equals("等待超时")) {
                    // 超时，发送请求回滚
                    System.out.println(entry.getKey() + " 发起回滚操作");
                    killDemo.rollback(entry.getKey());
                }
            } catch (Exception e) {
                // 捕获所有异常，包括超时异常
                // 建议：区分不同类型的异常，对TimeoutException单独处理
                e.printStackTrace();
                // 潜在问题：这里没有处理Future.get()抛出的TimeoutException
                // 应该添加对TimeoutException的特殊处理，可能需要回滚操作
            }
        });

        System.out.println("------- 库存操作日志 -------");
        // 统计并打印扣减成功的记录
        System.out.println(
                "扣减成功条数: " + killDemo.operateChangeLogList.stream().filter(e -> e.getOperateType().equals(1)).count());
        killDemo.operateChangeLogList.forEach(e -> {
            if (e.getOperateType().equals(1)) {
                System.out.println(e);
            }
        });

        // 统计并打印回滚的记录
        System.out.println(
                "扣减回滚条数: " + killDemo.operateChangeLogList.stream().filter(e -> e.getOperateType().equals(2)).count());
        killDemo.operateChangeLogList.forEach(e -> {
            if (e.getOperateType().equals(2)) {
                System.out.println(e);
            }
        });

        System.out.println("-------- 库存 --------");
        // 打印最终库存 - 注意这里显示的仍然是初始库存的值，应该修改为"最终库存数量"
        System.out.println("最终库存数量 :" + killDemo.stock);

        // 关闭线程池，释放资源
        executorService.shutdown();
        // 注意：这里没有等待线程池完全关闭，在实际应用中应该添加等待逻辑

    }

    /**
     * 回滚操作 - 对已扣减但请求超时的订单进行库存回滚
     * 该方法使用synchronized确保回滚操作的原子性，防止并发回滚导致库存不一致
     * 
     * @param userRequest 用户请求信息
     */
    private synchronized void rollback(UserRequest userRequest) {
        // 检查是否存在对应的扣减记录
        // 注意：这里使用stream查询可能不是最高效的方式，在大量数据时考虑使用索引查询
        if (operateChangeLogList.stream()
                .anyMatch(operateChangeLog -> operateChangeLog.getOrderId().equals(userRequest.getOrderId()) &&
                        operateChangeLog.getOperateType().equals(1))) {
            // 检查是否已经回滚过，避免重复回滚
            // 潜在问题：如果日志量大，这种遍历方式效率较低
            boolean hasRollback = operateChangeLogList.stream()
                    .anyMatch(operateChangeLog -> operateChangeLog.getOrderId().equals(userRequest.getOrderId()) &&
                            operateChangeLog.getOperateType().equals(2));
            if (hasRollback) {
                System.out.println("订单 " + userRequest.getOrderId() + " 已经回滚过，忽略重复回滚");
                return;
            }
            System.out.println("订单 " + userRequest.getOrderId() + " 执行库存回滚");
            // 增加库存
            stock += userRequest.getCount();
            // 记录回滚日志
            saveChangeLog(Lists.newArrayList(userRequest), 2);
        } else {
            // 没有找到对应的扣减记录，可能是请求未被处理
            System.out.println("订单 " + userRequest.getOrderId() + " 没有对应的扣减记录，无需回滚");
        }
    }

    // 模拟数据库库存记录
    // 使用volatile确保多线程可见性，但注意这不能保证复合操作的原子性
    // 对stock的修改操作需要在synchronized块中进行
    private volatile Integer stock = 6;

    // 请求队列，用于合并处理请求
    // 队列容量限制为10，超出时offer操作会阻塞或返回false
    // 注意：在高并发场景下，队列容量可能需要更大
    private BlockingQueue<RequestPromise> queue = new LinkedBlockingQueue<>(10);

    /**
     * 用户库存扣减请求处理
     * 该方法实现了请求合并和异步处理的核心逻辑：
     * 1. 将请求放入队列
     * 2. 等待合并处理线程处理并通知结果
     * 3. 处理超时情况
     * 
     * @param userRequest 用户请求信息，包含订单ID、用户ID和请求数量
     * @return 处理结果
     * @throws InterruptedException 当线程被中断时抛出
     */
    public Result operate(UserRequest userRequest) throws InterruptedException {
        // 快速失败：检查请求数量是否有效
        if (userRequest.getCount() <= 0) {
            return new Result(false, "请求数量必须大于0");
        }

        // 创建请求Promise对象，用于异步获取结果
        RequestPromise requestPromise = new RequestPromise(userRequest);
        synchronized (requestPromise) {
            // 将请求放入队列，设置入队超时时间为100毫秒
            // 潜在问题：如果队列已满，会等待100ms，可能导致客户端等待时间过长
            boolean enqueueSuccess = queue.offer(requestPromise, 100, TimeUnit.MILLISECONDS);
            if (!enqueueSuccess) {
                // 入队失败，返回系统繁忙
                return new Result(false, "系统繁忙，请稍后重试");
            }

            try {
                // 等待合并处理线程的通知，最多等待200毫秒
                // 注意：这里的wait会释放requestPromise对象的锁，允许其他线程调用notify
                requestPromise.wait(200);
                if (requestPromise.getResult() == null) {
                    // 超时未获得结果
                    // 潜在问题：此时请求可能仍在处理中，直接返回超时可能导致重复处理
                    return new Result(false, "等待超时");
                }
            } catch (InterruptedException e) {
                // 清理中断状态
                Thread.currentThread().interrupt();
                return new Result(false, "请求处理被中断");
            }
        }
        // 返回处理结果
        return requestPromise.getResult();
    }

    /**
     * 启动合并处理线程
     * 将多个请求合并为一个批次进行处理，提高系统吞吐量
     * 该方法实现了请求批处理的核心逻辑：
     * 1. 收集一定数量的请求或等待一定时间
     * 2. 批量处理请求
     * 3. 通知等待线程处理结果
     */
    public void mergeJob() {
        new Thread(() -> {
            // 存储待处理的请求
            List<RequestPromise> list = new ArrayList<>();
            while (true) {
                try {
                    // 队列为空时短暂休眠，避免CPU空转
                    // 注意：这种方式不是最优的，考虑使用带超时的poll操作代替
                    if (queue.isEmpty()) {
                        Thread.sleep(10);
                        continue;
                    }

                    // 批量处理的大小 - 每批最多处理3个请求
                    int batchSize = 3;
                    // 当前批次已收集的请求数
                    int collectedCount = 0;
                    // 设置最大等待时间，避免请求长时间等待
                    long maxWaitTime = 100; // 毫秒
                    long startTime = System.currentTimeMillis();

                    // 收集请求直到达到批处理大小或超过最大等待时间
                    while (collectedCount < batchSize) {
                        // 如果队列为空且已有请求，或者已超过最大等待时间，则处理当前批次
                        // 这确保了即使请求数量不足batchSize，也能在maxWaitTime时间内处理
                        if ((queue.isEmpty() && !list.isEmpty()) ||
                                (System.currentTimeMillis() - startTime > maxWaitTime && !list.isEmpty())) {
                            break;
                        }

                        // 从队列中取出请求，最多等待剩余的最大等待时间
                        // 计算剩余等待时间，确保总等待时间不超过maxWaitTime
                        long remainingWaitTime = Math.max(1, maxWaitTime - (System.currentTimeMillis() - startTime));
                        RequestPromise request = queue.poll(remainingWaitTime, TimeUnit.MILLISECONDS);
                        if (request != null) {
                            list.add(request);
                            collectedCount++;
                        }
                    }

                    // 如果没有收集到任何请求，继续下一轮循环
                    if (list.isEmpty()) {
                        continue;
                    }

                    // 模拟特定用户ID导致的处理延迟（仅用于演示超时场景）
                    // 当用户ID为5的请求在批次中时，整个批次会延迟处理
                    // 注意：这会导致批次中的所有请求都可能超时，而不仅仅是用户ID为5的请求
                    if (list.stream().anyMatch(e -> e.getUserRequest().getUserId().equals(5L))) {
                        Thread.sleep(200); // 故意延迟，触发超时
                    }

                    System.out.println(
                            Thread.currentThread().getName() + ":合并处理批次，请求数:" + list.size() + ", 请求详情:" + list);

                    // 计算当前批次的总请求数量
                    int sum = list.stream().mapToInt(e -> e.getUserRequest().getCount()).sum();

                    // 同步处理库存扣减，确保线程安全
                    // 使用synchronized块确保库存操作的原子性
                    synchronized (this) {
                        // 情况1：库存足够满足整个批次 - 批量处理
                        if (sum <= stock) {
                            // 开始事务 - 批量扣减库存
                            stock -= sum;
                            // 记录库存变更日志
                            saveChangeLog(
                                    list.stream().map(RequestPromise::getUserRequest).collect(Collectors.toList()), 1);
                            // 设置所有请求的处理结果
                            list.forEach(requestPromise -> {
                                requestPromise.setResult(new Result(true, "扣减成功"));
                                // 通知等待的线程
                                // 注意：这里对每个requestPromise对象加锁，确保notify操作的线程安全
                                synchronized (requestPromise) {
                                    requestPromise.notify();
                                }
                            });
                            list.clear();
                            continue;
                        }

                        // 情况2：库存不足满足整个批次，逐个处理
                        // 这种情况下，先到的请求优先得到处理
                        for (RequestPromise requestPromise : list) {
                            int count = requestPromise.getUserRequest().getCount();
                            if (count <= stock) {
                                // 库存足够，扣减并记录
                                stock -= count;
                                saveChangeLog(Lists.newArrayList(requestPromise.getUserRequest()), 1);
                                requestPromise.setResult(new Result(true, "扣减成功"));
                            } else {
                                // 库存不足
                                requestPromise.setResult(new Result(false, "库存不足"));
                            }
                            // 通知等待的线程
                            synchronized (requestPromise) {
                                requestPromise.notify();
                            }
                        }
                        list.clear();
                    }
                } catch (InterruptedException e) {
                    // 处理中断异常
                    System.err.println("合并处理线程被中断: " + e.getMessage());
                    // 对未处理的请求返回中断结果
                    // 确保所有等待的请求都能收到响应，避免客户端永久等待
                    list.forEach(requestPromise -> {
                        requestPromise.setResult(new Result(false, "系统处理异常"));
                        synchronized (requestPromise) {
                            requestPromise.notify();
                        }
                    });
                    list.clear();
                    // 恢复中断状态
                    // 注意：这里重新设置中断标志，允许外层代码检测到中断
                    Thread.currentThread().interrupt();
                }
                // 潜在问题：这里缺少对其他异常类型的处理
                // 建议添加catch(Exception e)块处理其他可能的异常，确保合并线程不会意外终止
            }
        }, "mergeThread").start();
    }

    // 模拟数据库操作日志表
    // order_id_operate_type 作为联合唯一索引
    // 使用线程安全的集合，确保多线程环境下的日志记录安全
    // 注意：在实际系统中，应该使用数据库事务确保库存和日志的一致性
    private final List<OperateChangeLog> operateChangeLogList = Collections.synchronizedList(new ArrayList<>());

    /**
     * 写库存操作流水记录
     * 该方法使用synchronized确保日志记录的原子性
     * 
     * @param list        用户请求列表
     * @param operateType 操作类型：1-扣减，2-回滚
     */
    private synchronized void saveChangeLog(List<UserRequest> list, int operateType) {
        // 将用户请求转换为操作日志记录
        // 注意：在实际系统中，这里应该是一个数据库事务，确保与库存操作的一致性
        List<OperateChangeLog> collect = list.stream()
                .map(userRequest -> new OperateChangeLog(
                        userRequest.getOrderId(),
                        userRequest.getCount(),
                        operateType))
                .collect(Collectors.toList());
        // 添加到日志列表
        // 虽然operateChangeLogList是线程安全的，但这里仍使用synchronized确保与其他方法的同步
        operateChangeLogList.addAll(collect);
    }
}

/**
 * 库存操作日志记录
 * 记录每次库存变更的详细信息
 * 包含订单ID、变更数量和操作类型（扣减/回滚）
 */
class OperateChangeLog {
    private Long orderId;
    private Integer count;
    // 1-扣减，2-回滚
    private Integer operateType;

    public OperateChangeLog(Long orderId, Integer count, Integer operateType) {
        this.orderId = orderId;
        this.count = count;
        this.operateType = operateType;
    }

    public Long getOrderId() {
        return orderId;
    }

    public void setOrderId(Long orderId) {
        this.orderId = orderId;
    }

    public Integer getCount() {
        return count;
    }

    public void setCount(Integer count) {
        this.count = count;
    }

    public Integer getOperateType() {
        return operateType;
    }

    public void setOperateType(Integer operateType) {
        this.operateType = operateType;
    }

    @Override
    public String toString() {
        return "OperateChangeLog{" +
                "orderId=" + orderId +
                ", count=" + count +
                ", operateType=" + operateType +
                '}';
    }
}

/**
 * 请求Promise对象
 * 用于异步处理请求并保存处理结果
 * 实现了请求与响应的异步解耦，允许批量处理请求
 */
class RequestPromise {
    private UserRequest userRequest;
    private Result result;

    public RequestPromise(UserRequest userRequest) {
        this.userRequest = userRequest;
    }

    public RequestPromise(UserRequest userRequest, Result result) {
        this.userRequest = userRequest;
        this.result = result;
    }

    public UserRequest getUserRequest() {
        return userRequest;
    }

    public void setUserRequest(UserRequest userRequest) {
        this.userRequest = userRequest;
    }

    public Result getResult() {
        return result;
    }

    public void setResult(Result result) {
        this.result = result;
    }

    @Override
    public String toString() {
        return "RequestPromise{" +
                "userRequest=" + userRequest +
                ", result=" + result +
                '}';
    }
}

/**
 * 请求处理结果
 * 包含处理是否成功的标志和消息
 * 用于向客户端返回处理状态和原因
 */
class Result {
    private Boolean success;
    private String msg;

    public Result(boolean success, String msg) {
        this.success = success;
        this.msg = msg;
    }

    public boolean isSuccess() {
        return success;
    }

    public void setSuccess(boolean success) {
        this.success = success;
    }

    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        this.msg = msg;
    }

    @Override
    public String toString() {
        return "Result{" +
                "success=" + success +
                ", msg='" + msg + '\'' +
                '}';
    }
}

/**
 * 用户请求信息
 * 包含订单ID、用户ID和请求数量
 * 作为秒杀请求的基本单元，包含了处理请求所需的全部信息
 */
class UserRequest {
    private Long orderId;
    private Long userId;
    private Integer count;

    public UserRequest(Long orderId, Long userId, Integer count) {
        this.orderId = orderId;
        this.userId = userId;
        this.count = count;
    }

    public Long getOrderId() {
        return orderId;
    }

    public void setOrderId(Long orderId) {
        this.orderId = orderId;
    }

    public Long getUserId() {
        return userId;
    }

    public void setUserId(Long userId) {
        this.userId = userId;
    }

    public Integer getCount() {
        return count;
    }

    public void setCount(Integer count) {
        this.count = count;
    }

    @Override
    public String toString() {
        return "UserRequest{" +
                "orderId=" + orderId +
                ", userId=" + userId +
                ", count=" + count +
                '}';
    }
}
